package cn.datawin.cmppserver.netway.handler;

import cn.datawin.cmppserver.netway.service.ITcpService;
import cn.datawin.cmppserver.netway.service.impl.CmppAPIServiceImpl;
import cn.datawin.cmppserver.util.CommonThreadPoolFactory;
import cn.datawin.cmppserver.util.LocalCache;
import cn.datawin.cmppserver.util.LongMsgCache;
import cn.datawin.sms.cmpp2.SmsDetailLog;
import cn.datawin.sms.enums.MessageType;
import cn.datawin.sms.enums.ProtocolType;
import cn.datawin.util.SMSUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.drondea.sms.channel.ChannelSession;
import com.drondea.sms.common.util.CommonUtil;
import com.drondea.sms.common.util.MsgId;
import com.drondea.sms.common.util.SystemClock;
import com.drondea.sms.message.ILongSMSMessage;
import com.drondea.sms.message.IMessage;
import com.drondea.sms.message.cmpp.CmppSubmitRequestMessage;
import com.drondea.sms.message.cmpp.CmppSubmitResponseMessage;
import com.drondea.sms.session.cmpp.CmppServerSession;
import com.drondea.sms.type.CmppConstants;
import com.drondea.sms.type.UserChannelConfig;
import com.google.common.cache.RemovalCause;
import com.google.common.cache.RemovalListener;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.EventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.net.InetSocketAddress;
import java.util.Date;

/**
 * cmpp服务器端接收到消息业务处理
 **/
@Slf4j
@ChannelHandler.Sharable
public class CmppServerBusinessHandler extends ChannelDuplexHandler {

    private static int ONE_HOUR = 1 * 60 * 60;

    private final RemovalListener<String, LongMsgCache> removalListener = notification -> {
        RemovalCause cause = notification.getCause();
        //超时的才算
        if (!RemovalCause.EXPIRED.equals(cause)) {
            return;
        }
        //以下为长短信组装超时发送回执的逻辑
        String key = notification.getKey();
        LongMsgCache msgCache = notification.getValue();
        //缓存满或者缓存组装超时回执情况
        String[] msgIds = msgCache.getMsgIds();
        String[] mobiles = msgCache.getMobiles();
        String subCode = msgCache.getSubCode();
        //key由协议、批次号、用户id组成
        String[] keys = key.split("_");
        String protocolType = keys[0];
        String batchNumber = keys[1];
        String account = keys[2];
        Long timestamp = msgCache.getTimestamp();
        log.error(StringUtils.join(mobiles, ",") + "长短信组装失败,msgIds:" + StringUtils.join(msgIds, ",") + "原因：" + cause);
        doLongMsgTimeout(account, mobiles, subCode, batchNumber, msgIds, protocolType, new Date(timestamp), cause.name());
    };

    /**
     * 保存长短信的每个片段的msgId
     */
    private LocalCache<LongMsgCache> msgIdCache = new LocalCache<>(ONE_HOUR, removalListener);

    /**
     * 缓存下游的msgId
     *
     * @return 是否已经缓存完成
     */
    protected boolean saveMsgId2Cache(ILongSMSMessage request, String msgId, String account, String[] mobiles,
                                      String subCode, long timestamp) {
        String localCacheKey = getService().genLocalCacheKey(request.getBatchNumber(), account);

        short pkNumber = request.getPkNumber();
        LongMsgCache msgCache = msgIdCache.get(localCacheKey);
        if (msgCache == null) {
            short pkTotal = request.getPkTotal();
            //第一次缓存注意线程安全
            msgCache = new LongMsgCache(pkTotal);
            LongMsgCache msgCachePre = msgIdCache.putIfAbsent(localCacheKey, msgCache);
            if (msgCachePre != null) {
                msgCache = msgCachePre;
            }
            msgCache.setMobiles(mobiles);
            msgCache.setSubCode(subCode);
            msgCache.setTimestamp(timestamp);
        }

        if (request.isMsgComplete()) {
            msgCache.setContent(request.getMsgContent());
        }
        return msgCache.saveCachedMsgId(pkNumber - 1, msgId);
    }

    protected LongMsgCache getLongMsgCache(ILongSMSMessage request, String account) {
        String localCacheKey = getService().genLocalCacheKey(request.getBatchNumber(), account);
        LongMsgCache msgCache = msgIdCache.get(localCacheKey);
        if (msgCache != null) {
            msgIdCache.remove(localCacheKey);
            return msgCache;
        }
        return null;
    }

    protected void doBizSubmit(ILongSMSMessage requestMessage, String account,
                               String[] mobiles, String subCode, final ChannelHandlerContext ctx,
                               String charset, String msgId, String protoType, Date inputDate) {
        if (StringUtils.isNotEmpty(subCode) && subCode.startsWith(account)) {
            subCode = StringUtils.replaceOnce(subCode, account, "");
        }

        String content;
        //获取响应给下游的msgId，用于report
        if (requestMessage.isLongMsg()) {
            LongMsgCache msgCache = getLongMsgCache(requestMessage, account);
            if (msgCache == null) {
                log.error(String.format("收集msgid失败====>%1$s", requestMessage));
                return;
            }
            msgId = msgCache.getMsgIdStr();
            content = msgCache.getContent();
        } else {
            content = requestMessage.getMsgContent();
        }

        InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = socket.getAddress().getHostAddress();

        // 长短信合并后的批次号
        CmppSubmitRequestMessage cmppSubmitRequestMessage = (CmppSubmitRequestMessage) requestMessage;
        String batchNumber = requestMessage.getBatchNumber();

        String userMsgId = cmppSubmitRequestMessage.getMsgId().toString();
        int userSequence = cmppSubmitRequestMessage.getSequenceId();
        short userPktotal = cmppSubmitRequestMessage.getPkTotal();
        short userPkNumber = cmppSubmitRequestMessage.getPkNumber();
        short userTpudhi = cmppSubmitRequestMessage.getTpUdhi();
        short userMsgLen = cmppSubmitRequestMessage.getMsgLength();
        log.debug("{}-{}-{}-{}-{}-{}", userMsgId, userSequence, userPktotal, userPkNumber, userTpudhi, userMsgLen);

        try {
            String mobileCompose = StringUtils.join(mobiles, ",");
            // 保存input 并通知分拣
            SmsDetailLog smsDetailLog = new SmsDetailLog();
            smsDetailLog.setProtocolTypeCode(protoType);
            smsDetailLog.setMessageTypeCode(MessageType.SMS.toString());
            smsDetailLog.setSourceIp(clientIp);
            //input.客户id
            smsDetailLog.setInputDate(inputDate);
            smsDetailLog.setPhoneNos(mobileCompose);
            smsDetailLog.setCountryCode("86");

            smsDetailLog.setContent(content);
            smsDetailLog.setSubCode(subCode);
            smsDetailLog.setCharset(charset);
            smsDetailLog.setMsgBatchNo(batchNumber);
            smsDetailLog.setCrMsgId(msgId);
            smsDetailLog.setCstmCustomerCode(account);

            smsDetailLog.setUserMsgId(userMsgId);
            smsDetailLog.setUserSequence(userSequence);
            smsDetailLog.setUserPkTotal(userPktotal);
            smsDetailLog.setUserPkNumber(userPkNumber);
            smsDetailLog.setUserTpudhi(userTpudhi);
            smsDetailLog.setUserMsgLen(userMsgLen);
            getService().saveInput(smsDetailLog);
        } catch (Exception e) {
            log.warn(String.format("response已响应：0，业务检查失败====>%1$s", requestMessage.toString()));
            log.error(e.getMessage(), e);
        }

    }

//    private int doSubmits(String account, String[] mobiles, String spNo, String content, String batchNumber,
//                          String clientIp, String charset, String msgId, String protocolType, Date inputDate) {
//        try {
//            Input input = buildInput(account, mobiles, spNo, content, batchNumber, clientIp, charset, msgId, protocolType, inputDate);
//            return getService().saveInput(input);
//        } catch (Exception e) {
//            log.error(e.getMessage(), e);
//            return -1;
//        }
//    }

    /**
     * 长短信组装超时使用
     */
    private void doLongMsgTimeout(String account, String[] mobiles, String spNo, String batchNumber,
                                  String[] msgIds, String protocolType, Date inputDate, String cause) {
        //Input input = buildInput(account, mobiles, spNo, "", batchNumber, "127.0.0.1", "UCS2", "", protocolType, inputDate);
        log.error("account={},mobiles={},spNo={},msgIds={},protocolType={},inputDate={},系统错误内容不完整,cause={}", account, mobiles, spNo, msgIds, protocolType, inputDate, cause);
    }

//    private Input buildInput(String account, String[] mobiles, String spNo, String content, String batchNumber,
//                             String clientIp, String charset, String msgId, String protocolType, Date inputDate) {
//        String mobileCompose = StringUtils.join(mobiles, ",");
//        // 保存input 并通知分拣
//        Input input = new Input();
//        input.setProtocolTypeCode(protocolType);
//        input.setMessageTypeCode(MessageType.SMS.toString());
//        input.setSourceIP(clientIp);
//        //input.客户id
//        input.setInputDate(inputDate);
//        input.setPhoneNos(mobileCompose);
//        input.setCountryCode("86");
//
//        input.setContent(content);
//        input.setSubCode(spNo);
//        input.setCharset(charset);
//        input.setMsgBatchNo(batchNumber);
//        input.setCrMsgId(msgId);
//        input.setAccount(account);
//        return input;
//    }

    /**
     * 获取service实现
     */

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof CmppSubmitRequestMessage) {
            CmppSubmitRequestMessage requestMessage = (CmppSubmitRequestMessage) msg;
            ChannelSession channelSession = CommonUtil.getChannelSession(ctx.channel());
            CmppSubmitResponseMessage responseMessage = new CmppSubmitResponseMessage(requestMessage.getHeader());
            //msgid 响应，回执要用
            responseMessage.setMsgId(new MsgId());
            //用户提交时间
            long timestamp = SystemClock.now();

            //根据连接获取用户id
            CmppServerSession cmppServerSession = (CmppServerSession) channelSession;
            UserChannelConfig userChannelConfig = cmppServerSession.getUserChannelConfig();
            String account = userChannelConfig.getId();
            //响应
            sendMessage(channelSession, responseMessage);

            short version = userChannelConfig.getVersion();
            String versionStr = getCmppVersion(version);

            CommonThreadPoolFactory.getInstance().getBizPoolExecutor().submit(() -> {
                String msgId = responseMessage.getMsgId().toString();
                String[] mobiles = requestMessage.getDestTerminalId();
                String subCode = requestMessage.getSrcId();

                boolean isCacheAll = true;
                if (requestMessage.isLongMsg()) {
                    isCacheAll = saveMsgId2Cache(requestMessage, msgId, account, mobiles, subCode, timestamp);
                }

                if (isCacheAll) {
                    String charset = SMSUtil.getCharsetByByte(requestMessage.getMsgFmt().getValue());
                    doBizSubmit(requestMessage, account, mobiles, subCode, ctx, charset, msgId, versionStr, new Date(timestamp));
                }
            });
        }
        super.channelRead(ctx, msg);
    }


    /**
     * 放到netty中发送，在用户线程中发送有问题
     *
     * @param channelSession
     * @param iMessage
     */
    void sendMessage(ChannelSession channelSession, IMessage iMessage) {
        if (!channelSession.getChannel().isActive()) {
            return;
        }
        EventExecutor executor = channelSession.getChannel().pipeline().firstContext().executor();
        executor.submit(() -> {
            channelSession.sendMessage(iMessage);
        });
    }
    private String getCmppVersion(short version) {
        if (version < CmppConstants.VERSION_30) {
            return ProtocolType.CMPP2.toString();
        }
        return ProtocolType.CMPP3.toString();
    }

    protected ITcpService getService() {
        return SpringUtil.getBean(CmppAPIServiceImpl.class);
    }
}
